package org.pentaho.kettle.plugin.mqtt.steps.producer;


import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.repository.Repository;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.metastore.api.IMetaStore;

public class ProducerStep extends BaseStep implements StepInterface {
    private static final Class<?> PKG = ProducerStep.class;
    private MemoryPersistence persistence = new MemoryPersistence();
    private MqttClient mqttClientProducer;
    private MqttConnectOptions connOpts;
    private MqttMessage msg;
    private String message;

    public ProducerStep(StepMeta s, StepDataInterface stepDataInterface, int c, TransMeta t, Trans dis) {
        super(s, stepDataInterface, c, t, dis);
    }

    public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
        ProducerStepMeta meta = (ProducerStepMeta)smi;
        ProducerStepData data = (ProducerStepData)sdi;
        return super.init(meta, data);
    }

    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
        ProducerStepMeta meta = (ProducerStepMeta)smi;
        ProducerStepData data = (ProducerStepData)sdi;
        Object[] r = this.getRow();
        String topic = meta.getPubTopic();
        String clientId = meta.getPubClientId();
        String broker = meta.getPubBroker();
        int outQos = meta.getPubQos();
        if (r == null) {
            this.setOutputDone();

            try {
                this.mqttClientProducer.disconnect();
                this.logBasic("Disconnected to: " + broker + " & Topic: " + topic);
                return false;
            } catch (MqttException var11) {
                this.logErrorMsg(var11.getReasonCode(), var11.getMessage(), var11.getLocalizedMessage(), var11.getCause(), var11);
                var11.printStackTrace();
                return false;
            }
        } else {
            if (this.first) {
                this.first = false;

                try {
                    this.mqttClientProducer = new MqttClient(broker, clientId, this.persistence);
                    this.connOpts = new MqttConnectOptions();
                    this.connOpts.setCleanSession(true);
                    this.mqttClientProducer.connect(this.connOpts);
                    this.logBasic("Connected to: " + broker + " & Topic: " + topic);
                } catch (MqttException var13) {
                    this.logErrorMsg(var13.getReasonCode(), var13.getMessage(), var13.getLocalizedMessage(), var13.getCause(), var13);
                    var13.printStackTrace();
                }

                data.outputRowMeta = this.getInputRowMeta().clone();
                meta.getFields(data.outputRowMeta, this.getStepname(), (RowMetaInterface[])null, (StepMeta)null, this, (Repository)null, (IMetaStore)null);
            }

            try {
                int index = this.getInputRowMeta().indexOfValue(meta.getPubMessage().toString());
                this.message = r[index].toString();
                this.msg = new MqttMessage(this.message.getBytes());
                this.msg.setQos(outQos);
                this.mqttClientProducer.publish(topic, this.msg);
            } catch (MqttException var12) {
                this.logErrorMsg(var12.getReasonCode(), var12.getMessage(), var12.getLocalizedMessage(), var12.getCause(), var12);
                var12.printStackTrace();
                return false;
            }

            this.putRow(data.outputRowMeta, r);
            if (this.checkFeedback(this.getLinesRead())) {
                this.logBasic("Linenr " + this.getLinesRead());
            }

            return true;
        }
    }

    public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
        ProducerStepMeta meta = (ProducerStepMeta)smi;
        ProducerStepData data = (ProducerStepData)sdi;
        super.dispose(meta, data);
    }

    private void logErrorMsg(int reason, String message, String locMessage, Throwable cause, MqttException ex) {
    }
}
